<?php

namespace Php\Queue\work;

use Php\Queue\RedisDriver;

class Liunx extends QueueMaster
{
	public $query;
	/**
	 * 子进程数量
	 *
	 * @var int
	 */
	public $num;
	
	/**
	 * 记录子进程pid
	 *
	 * @var array
	 */
	public $workers = [];
	
	public function __construct($query)
	{
		$this->query = $query;
		$this->query->isSoole = false;
		
		$fp = popen("cat /proc/cpuinfo | grep 'cpu cores'|uniq",'r');
		$rs = fread($fp,1024);
		pclose($fp);
		$rs = explode(':', $rs);
		$this->num = (int)trim($rs[1]);
		$this->num = 20;
	}
	
	public function run()
	{
		$this->createChildProcess();
		$this->monitor();
	}
	
	private function createChildProcess()
	{
		while (count($this->workers) < $this->num) {
			$pid = \pcntl_fork();
			if ($pid > 0) {
				$this->workers[$pid] = $pid;
			} elseif ($pid == 0) {
				$childPid = getmypid();
				echo "子进程[已创建, pid = $childPid\n";
				$this->workers();
	
				exit();
			} else {
				echo "Fork子进程失败\n";
			}
		}
	}
	
	private function workers()
	{
		$num = 0;
		$redis = RedisDriver::connect($this->query->config, 0);
		while ($num <= 500) {
			// 将达到指定时间的消息加入到队列.
			$this->query->migrateExpiredJobs($redis, 100);
			$result = $this->query->pop($redis);
			if ($result) {
				$class = $result['job'];
				unset($result['job']);
				$class = new $class;
				$class->job($this->query, $result);
				$num++;
			} else {
				sleep(2);
				// 当队列中没有消息时,将没有标记成功的消息恢复到队列中
				$this->query->recoverMessage($redis, 100);
			}
		}
	}
	
	private function monitor()
	{
		while (($pid = pcntl_wait($status)) > 0) {
			unset($this->workers[$pid]);
			$this->createChildProcess();
		}
	}
	
	
}